package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.VisitorStats;
import com.atguigu.gmall.realtime.utils.ClickhouseUtils;
import com.atguigu.gmall.realtime.utils.DateUtils;
import com.atguigu.gmall.realtime.utils.KafkaUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

/**
 * 访客主题统计
 *
 * @author lvbingbing
 * @date 2022-06-30 10:04
 */
public class VisitStatsApp {
    public static void main(String[] args) throws Exception {
        // 1. 基本环境准备
        // 1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 并行度设置
        env.setParallelism(4);
        // 2. 检查点相关设置
        // 2.1 开启检查点
        env.enableCheckpointing(10000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置检查点超时时间
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointTimeout(60000L);
        // 2.3 开启外部化检查点，作业取消时保留检查点
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 6000L));
        // 2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/ck/gmall"));
        // 2.6 指定操作 hdfs 用户
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        // 3. 从 kafka 读取数据，做结构转换，并指定时间戳和 watermark
        String groupId = "visit_stats_app_group";
        String pvTopic = "dwd_page_log";
        String uvTopic = "dwm_unique_visitor";
        String ujdTopic = "dwm_user_jump_detail";
        // 3.1 获取 PV 主题中的数据并做转换
        SingleOutputStreamOperator<VisitorStats> pvDs = env.addSource(KafkaUtils.getKafkaSource(pvTopic, groupId))
                .map(s -> {
                    JSONObject jsonObj = JSON.parseObject(s);
                    JSONObject commonObj = jsonObj.getJSONObject("common");
                    JSONObject pageObj = jsonObj.getJSONObject("page");
                    return VisitorStats.builder()
                            // 版本
                            .vc(commonObj.getString("vc"))
                            // 渠道
                            .ch(commonObj.getString("ch"))
                            // 地区
                            .ar(commonObj.getString("ar"))
                            // 新老用户标识
                            .is_new(commonObj.getString("is_new"))
                            .pv_ct(1L)
                            .uv_ct(0L)
                            .sv_ct(StringUtils.isEmpty(pageObj.getString("last_page_id")) ? 1L : 0L)
                            .uj_ct(0L)
                            .dur_sum(pageObj.getLong("during_time"))
                            .ts(jsonObj.getLong("ts"))
                            .build();
                });

        // 3.2 获取独立访客主题中的数据并做转换
        SingleOutputStreamOperator<VisitorStats> uvDs = env.addSource(KafkaUtils.getKafkaSource(uvTopic, groupId))
                .map(s -> {
                    JSONObject jsonObj = JSON.parseObject(s);
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    return VisitorStats.builder()
                            .vc(commonJsonObj.getString("vc"))
                            .ch(commonJsonObj.getString("ch"))
                            .ar(commonJsonObj.getString("ar"))
                            .is_new(commonJsonObj.getString("is_new"))
                            .pv_ct(0L)
                            .uv_ct(1L)
                            .sv_ct(0L)
                            .uj_ct(0L)
                            .dur_sum(0L)
                            .ts(jsonObj.getLong("ts"))
                            .build();
                });
        // 3.3 获取用户跳出明细主题中的数据并做转换
        SingleOutputStreamOperator<VisitorStats> ujdDs = env.addSource(KafkaUtils.getKafkaSource(ujdTopic, groupId))
                .map(s -> {
                    JSONObject jsonObj = JSON.parseObject(s);
                    JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                    return VisitorStats.builder()
                            .vc(commonJsonObj.getString("vc"))
                            .ch(commonJsonObj.getString("ch"))
                            .ar(commonJsonObj.getString("ar"))
                            .is_new(commonJsonObj.getString("is_new"))
                            .pv_ct(0L)
                            .uv_ct(0L)
                            .sv_ct(0L)
                            .uj_ct(1L)
                            .dur_sum(0L)
                            .ts(jsonObj.getLong("ts"))
                            .build();
                });

        // 4. 多流 union，合并之后指定 watermark 和 事件时间字段
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithWatermarkDs = pvDs.union(uvDs, ujdDs)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
                    @Override
                    public long extractTimestamp(VisitorStats visitorStats, long recordTimestamp) {
                        return visitorStats.getTs();
                    }
                }));
        // 5. 按照维度对数据进行分组，分组之后开窗，开窗之后进行聚合计算
        SingleOutputStreamOperator<VisitorStats> aggregateVisitorStats = visitorStatsWithWatermarkDs.keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VisitorStats visitorStats) throws Exception {
                        return Tuple4.of(visitorStats.getVc(), visitorStats.getCh(), visitorStats.getAr(), visitorStats.getIs_new());
                    }
                })
                // 开一个10s的滚动事件时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 聚合计算
                .reduce(new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats visitorStats1, VisitorStats visitorStats2) throws Exception {
                        visitorStats1.setPv_ct(visitorStats1.getPv_ct() + visitorStats2.getPv_ct());
                        visitorStats1.setUv_ct(visitorStats1.getUv_ct() + visitorStats2.getUv_ct());
                        visitorStats1.setSv_ct(visitorStats1.getSv_ct() + visitorStats2.getSv_ct());
                        visitorStats1.setUj_ct(visitorStats1.getUj_ct() + visitorStats2.getUj_ct());
                        return visitorStats1;
                    }
                }, new WindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> tuple4, TimeWindow window, Iterable<VisitorStats> visitorStatsIterable, Collector<VisitorStats> out) throws Exception {
                        // 设置窗口起止时间
                        for (VisitorStats visitorStats : visitorStatsIterable) {
                            visitorStats.setStt(DateUtils.formatDate(new Date(window.getStart())));
                            visitorStats.setEdt(DateUtils.formatDate(new Date(window.getEnd())));
                            visitorStats.setTs(System.currentTimeMillis());
                            // 将处理之后的数据发送到下游
                            out.collect(visitorStats);
                        }
                    }
                });
        aggregateVisitorStats.print("访客主题");
        // 6. 将轻度聚合的数据写到 clickhouse 中
        String executeSql = "insert into visitor_stats_2021 values(?,?,?,?,?,?,?,?,?,?,?,?)";
        aggregateVisitorStats.addSink(ClickhouseUtils.operateClickhouseJdbcSink(executeSql));
        // 7. 触发程序执行
        env.execute();
    }
}

























